package com.atguigu.dga.governance.assessor.calc;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.dga.common.HttpUtil;
import com.atguigu.dga.common.MetaConst;
import com.atguigu.dga.ds.bean.TDsTaskInstance;
import com.atguigu.dga.governance.assessor.Assessor;
import com.atguigu.dga.governance.bean.AssessParam;
import com.atguigu.dga.governance.bean.GovernanceAssessDetail;
import com.atguigu.dga.governance.bean.GovernanceMetric;
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

@Component("DATA_SKEW")
public class DataSkewAssessor extends Assessor {

    @Value("${spark-history-server.url}")
    String historyUrl;

    //    1 找到yarn_id    通过表名找到ds的task_instance 其中的字段app_link  就是yarn_id
//        2 通过yarn_id 获得整个任务的信息
//        http://hadoop102:18080/api/v1/applications/application_1682967712307_0019
//        提取成功的attempt_id
//       3 获取该次attempt 的stage列表
//        http://hadoop102:18080/api/v1/applications/{yarn_id}/{attempt_id}/stages
//
//       4  每个stage的详细信息  主要是 其下的task情况
//        http://hadoop102:18080/api/v1/applications/{yarn_id}/{attempt_id}/stages/{stage_id}
//       5  获得其下的json   提取 最大 task duration 和 其他的平均值进行 差值   超过的部分的百分比 和 参考百分比进行比对
    @Override
    protected void assessProblem(GovernanceAssessDetail governanceAssessDetail, AssessParam assessParam) throws Exception {
        //    1 找到yarn_id    通过表名找到ds的task_instance 其中的字段app_link  就是yarn_id
        List<TDsTaskInstance> dsTaskInstanceList = assessParam.getTDsTaskInstanceList();
        if(dsTaskInstanceList==null||dsTaskInstanceList.size()==0){
            return;
        }
        TDsTaskInstance  taskInstanceSuccess=null;
        for (TDsTaskInstance tDsTaskInstance : dsTaskInstanceList) {
            if(tDsTaskInstance.getState().toString().equals(MetaConst.TASK_STATE_SUCCESS.toString())){
                taskInstanceSuccess=tDsTaskInstance;
            }
        }
        String yarnId = taskInstanceSuccess.getAppLink();
        if(yarnId==null||yarnId.length()==0){
            return;
        }



//        2 通过yarn_id 获得整个任务的信息
//        http://hadoop102:18080/api/v1/applications/application_1682967712307_0019
//        提取成功的attempt_id
        String attemptId  = getAttemptIdByYarnId(yarnId);


//       3 获取该次attempt 的stage列表
//        http://hadoop102:18080/api/v1/applications/{yarn_id}/{attempt_id}/stages
        List<String> stageIdList=   getStageListByAttemptId(yarnId,attemptId);
//
//       4  每个stage的详细信息(每个stage总 时长，最大任务时长，其他任务的平均时长 ，任务数 )  主要是 其下的task情况
//        http://hadoop102:18080/api/v1/applications/{yarn_id}/{attempt_id}/stages/{stage_id}
        List<StageInfo> stageInfoList= getStageInfoList(yarnId,attemptId,stageIdList);

//       5  获得其下的json   提取 最大 task duration 和 其他的平均值进行 差值   超过的部分的百分比 和 参考百分比进行比对
        GovernanceMetric governanceMetric = assessParam.getGovernanceMetric();
        String metricParamsJson = governanceMetric.getMetricParamsJson();
        JSONObject paramJsonObj = JSON.parseObject(metricParamsJson);
        Integer paramPercent = paramJsonObj.getInteger("percent");
        Integer stageDurSeconds = paramJsonObj.getInteger("stage_dur_seconds");

        for (StageInfo stageInfo : stageInfoList) {
            if((stageInfo.stageDurationMs/1000) >stageDurSeconds){
                Integer curStagePercent=(stageInfo.maxTaskDurationMs - stageInfo.otherTaskAvgDurationMs)*100/ stageInfo.otherTaskAvgDurationMs;
                if(curStagePercent>paramPercent){
                    governanceAssessDetail.setAssessScore(BigDecimal.ZERO);
                    governanceAssessDetail.setAssessProblem("存在数据倾斜： 其中最大任务时间："+stageInfo.maxTaskDurationMs +"其他任务平均时间："+stageInfo.otherTaskAvgDurationMs);
                    return;
                }
            }
        }
    }

    private List<StageInfo> getStageInfoList(String yarnId, String attemptId, List<String> stageIdList) {
        List<StageInfo> stageInfoList=new ArrayList<>();
        for (String stageId : stageIdList) {
            String stageUrl = historyUrl + "/" + yarnId + "/" + attemptId + "/stages/"+stageId;
            String stageJson = HttpUtil.get(stageUrl);
            List<JSONObject> stageJsonObjList = JSON.parseArray(stageJson, JSONObject.class);

            for (JSONObject stageJsonObj : stageJsonObjList) {
                if(stageJsonObj.getString("status").equals("COMPLETE")){

                    //一个stage
                    StageInfo stageInfo = new StageInfo();
                    stageInfo.setTaskNum(stageJsonObj.getInteger("numCompleteTasks"));
                    JSONObject tasksJsonObj = stageJsonObj.getJSONObject("tasks");
                    for (Object taskObj : tasksJsonObj.values()) {

                        JSONObject taskJsonObj = (JSONObject) taskObj;
                        if(taskJsonObj.getString("status").equals("SUCCESS")){
                            Integer duration = taskJsonObj.getInteger("duration");
                            stageInfo.setMaxTaskDurationMs(Math.max(stageInfo.getMaxTaskDurationMs(),duration));
                            stageInfo.setTaskSumDurationMs( stageInfo.getTaskSumDurationMs()+duration );
                        }

                    }
                    if(stageInfo.taskNum>1){
                        stageInfo.setOtherTaskAvgDurationMs((stageInfo.taskSumDurationMs-stageInfo.maxTaskDurationMs)/(stageInfo.taskNum-1) );

                    }else{
                        stageInfo.setOtherTaskAvgDurationMs(stageInfo.taskSumDurationMs);
                    }
                   stageInfo.setStageDurationMs(stageInfo.getMaxTaskDurationMs()); //认定： stage耗时 约等于最长任务的耗时
                    stageInfoList.add(stageInfo);
                }

            }


        }
        return stageInfoList;
    }

    private List<String> getStageListByAttemptId(String yarnId,String attemptId) {
        String stagesUrl = historyUrl + "/" + yarnId + "/" + attemptId + "/stages";
        String stagesJson = HttpUtil.get(stagesUrl);
        List<JSONObject> stageJsonobjList = JSON.parseArray(stagesJson, JSONObject.class);
        List<String> stageIdList = stageJsonobjList.stream().filter(jsonObject -> jsonObject.getString("status").equals("COMPLETE"))
                .map(jsonObject -> jsonObject.getString("stageId")).collect(Collectors.toList());

        return stageIdList;
    }

    public String getAttemptIdByYarnId(String yarnId){
        String attemptsUrl = historyUrl + "/" + yarnId;
        String appJson = HttpUtil.get(attemptsUrl);
        String completeAttemptId=null;
        JSONObject appJsonObj = JSON.parseObject(appJson);

        JSONArray attemptJsonArr = appJsonObj.getJSONArray("attempts");
        for (int i = 0; i < attemptJsonArr.size(); i++) {
            JSONObject attemptJsonObj = attemptJsonArr.getJSONObject(i);
            if(attemptJsonObj.getBoolean("completed")){
                 return attemptJsonObj.getString("attemptId");
            }
        }
        throw  new RuntimeException("未找到完成的attempt");

    }

    //stage的详细信息(每个stage总 时长，最大任务时长，其他任务的平均时长 ，任务数 )
    @Data
    class StageInfo{
        Integer stageDurationMs=0;
        Integer maxTaskDurationMs=0;
        Integer taskSumDurationMs=0;
        Integer otherTaskAvgDurationMs=0;

        Integer taskNum=0;

    }
}
